package com.wfg.bigdata.spark.sparksql;

import org.apache.commons.lang.StringUtils;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.api.java.UDF1;
import org.apache.spark.sql.types.IntegerType$;

/**
 * @author wufagang
 * @description
 * @date 2024年05月09日 23:01
 */
public class SparkSQL04_Source_CSV_Case_2 {
    public static void main(String[] args) {
        SparkSession sparkSession = SparkSession.builder().appName("sparkSql").master("local[*]").getOrCreate();
        Dataset<Row> csv = sparkSession.read()
                .option("encoding", "UTF-8")
                .option("multiline", "true") // 单元格内存在回车符
                .option("escape", "\\")  // 假设使用反斜杠作为转义字符
                .option("header", "true")
                .csv("datas/sql/product.csv");
        // TODO 执行逻辑操作
        sparkSession.udf().register("prefixName", new UDF1<String, Integer>() {
            @Override
            public Integer call(String s) throws Exception {
                if(StringUtils.isEmpty(s)){
                    return 0;
                }
                return s.split("-").length;
            }
        }, IntegerType$.MODULE$);

//        csv.foreach(
//            row -> {
//                System.out.println(row.get(2));
//                System.out.println(row.get(1));
//            }
//        );


        csv.write()
                .option("encoding","UTF-8")
                .option("multiline", "true")
                .option("escape", "\\")  // 假设使用反斜杠作为转义字符
                .mode("append")
                .option("header", "true")
                .csv("output4");

        sparkSession.close();
    }
}
